package t20240424;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Test {


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);


        String sourceSql = "CREATE TABLE datagen (" +
                "    stime INT," +
                "    sn STRING," +
                "    k1v1 STRING," +
                "    k1v2 STRING," +
                "    k1v3 STRING," +
                "    k2v1 STRING," +
                "    k2v2 STRING," +
                "    ts AS localtimestamp," +
                "    WATERMARK FOR ts AS ts" +
                "  ) WITH (" +
                "    'connector' = 'datagen'," +
                "    'rows-per-second'='1'," +
                "    'fields.stime.kind'='sequence'," +
                "    'fields.stime.start'='1'," +
                "    'fields.stime.end'='10000'," +
                "    'fields.sn.length'='1'," +
                "    'fields.k1v1.length'='3'," +
                "    'fields.k1v2.length'='3'," +
                "    'fields.k1v3.length'='3'," +
                "    'fields.k2v1.length'='3'," +
                "    'fields.k2v2.length'='3'" +
                "  )";

        String sinkSql =
                "  CREATE TABLE print_table (" +
                "    stime INT," +
                "    sn STRING," +
                "    k1v1 STRING," +
                "    k1v2 STRING," +
                "    k1v3 STRING," +
                "    k2v1 STRING," +
                "    k2v2 STRING," +
                "    ts timestamp(3)," +
                "    kv STRING" +
                "    ) WITH (" +
                "    'connector' = 'print'" +
                "  )";

        String execSql = "INSERT INTO print_table select stime,sn,k1v1,k1v2,k1v3,k2v2,k2v2,ts," +
                " JSON_OBJECT('name',\"jj\", \"age\",18) as kv " +
                " from datagen";

        tableEnvironment.executeSql(sourceSql);
        System.out.println("-----1");
        tableEnvironment.executeSql(sinkSql);
        System.out.println("-----2");
        tableEnvironment.executeSql(execSql);


    }

}
